// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "sendthread.h"

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

#include <sstream>

#include "common.h"
#include "pigasus-types.h"
#include "serialization.h"
#include "threadpool.h"

using namespace std;

extern ThreadPool *send_thread_pool;
extern pthread_mutex_t *mutex_send_thread_pool;

SendThread::SendThread(Map &server_, List &files_, List &inputs_) {
  type = kSendThreadType;
  server = &server_;
  files = &files_;
  inputs = &inputs_;
}

SendThread::~SendThread() {}

void SendThread::Run() {
  FILE *p;
  char ignore[20];
  char *err;
  string buffer;
  string user = (*(string*)(server->Get("user")));
  debug("user is %s\n", user.c_str());
  string host = (*(string*)(server->Get("host")));
  
  debug("SendThread %d: Starting (user: %s, host: %s).\n", id, user.c_str(),
        host.c_str());
  
  // Create pigasus directory  
  debug("SendThread %d: Creating remote directory for Pigasus use.\n", id);
  buffer = "ssh -i pig_rsa -q " + user + "@" + host + " mkdir -p pig-"
    + IntToString(id);
  p = popen(buffer.c_str(), "r");
  while (!feof(p)) err = fgets(ignore, sizeof(ignore), p);
  pclose(p);

  // Serialize inputs.
  string serialized_inputs = Serialize(*inputs);
  string serialized_filename = "serialized-inputs-" + IntToString(id) + ".dat";
  if (!WriteStringToFile(serialized_filename, serialized_inputs)) {
    // TODO: Handle error. Problem writing file.
    debug("SendThread %d: Error writing serialized inputs file %s.\n",
          id, serialized_filename.c_str());
  }
  
  // Format string to SCP files over
  buffer = "scp -i pig_rsa runjob.sh " + serialized_filename;
  debug("There are %d files to send\n", files->GetSize());
  for (int i = 0; i < files->GetSize(); ++i) {
    buffer += " " + (*((string*)files->Get(i)));
  }
  for (int i = 0; i < inputs->GetSize(); ++i) {
    if (inputs->GetType(i) == kFileType) {
      buffer += " " + ((File*)(inputs->Get(i)))->Filename();
    }
  }

  // TODO: add the pigasus job executable...
  // I think this should be done in the JobThread? --John

  // SCP files
  debug("SendThread %d: Sending files...\n", id);
  buffer = buffer + " " + user + "@" + host + ":pig-" + IntToString(id);
  debug("SendThread %d: %d files and %d inputs.\n", id, files->GetSize(),
        inputs->GetSize());
  debug("SendThread %d: buffer: %s.\n", id, buffer.c_str());
  p = popen(buffer.c_str(), "r");
  if (!p) {
    debug("ERROR: cannot open pipe\n");
  }
  while (!feof(p)) {
    err = fgets(ignore, sizeof(ignore), p);
    debug("SendThread %d: ignore: %s\n", id, err);
  }
  pclose(p);

  // Delete local copy of the serialized inputs.
  if (remove(serialized_filename.c_str()) != 0) {
    // TODO: Handle error. Problem deleting serialized inputs file.
    debug("SendThread %d: Error deleting serialized inputs %s.\n", 
          id, serialized_filename.c_str());
  }
  
  // Start execution
  string job_executable = *((string*)files->Get(0));
  buffer = "./push.sh " + user + " " + host + " " + IntToString(id) + " "
    + job_executable + " " + serialized_filename;
  debug("Buffer is %s\n", buffer.c_str());
  p = popen(buffer.c_str(), "r");
  while (!feof(p)) err = fgets(ignore, sizeof(ignore), p);
  pclose(p);
  
  // Set status to done  
  pthread_mutex_lock(mutex_send_thread_pool);
  send_thread_pool->SetStatus(id, kThreadStatusMarkAsDone);
  pthread_mutex_unlock(mutex_send_thread_pool);
}
